#include "modules/synchronizer.h"
#include <atomic>

namespace mod {
	SyncModule::SyncModule(std::shared_ptr<Context> ctx, const std::string &module_name) //
	    : Module(ctx, module_name), running(false), sockfd(-1) {}

	bool SyncModule::Init() {
		// 1. 初始化套接字(共享网关套接字)
		auto listener = this->ctx->getListenerByPort(this->ctx->config.Gateway.GatewayPort);
		if (listener) {
			if (listener->sockfd < 1) {
				log_fatal("[Sync] sockfd error: %d .", listener->sockfd);
				return false;
			}
			sockfd = listener->sockfd;
		}

		// 2. 初始化Agent模块,用于动态创建Agent服务
		// agent = (AgentModule *) ctx->Mod(MODULE_AGENT);
		// if (!agent) {
		// 	log_fatal("[Sync] agent is null.");
		// 	return false;
		// }

		// 3. 同步进程
		// thread_sync.reset(new thread(&SyncModule::sync_handler, this));

		// 4. 订阅客户端代理创建事件,记录Nat路由信息。
		ctx->eventBus.Subscribe<AgentedEvent>([this](const AgentedEvent &event) {
			ctx->SetNatP2SMapper(event.sockFd, event.mate);           // Fd => NatMate
			ctx->SetNatS2PMapper(event.mate.hashKey(), event.sockFd); // NatMate => Fd
		});

		return true;
	}

	void SyncModule::Start() {
		if (running) return; // 避免重复启动
		running     = true;
		thread_sync = std::thread(&SyncModule::sync_handler, this);
	}

	// 同步线程处理程序,用于同步服务信息到其他网关中
	void SyncModule::sync_handler() const {
		// 等待第一次召唤完成
		this_thread::sleep_for(chrono::milliseconds(this->ctx->config.Multicast.Interval + 200));

		int sync_count = 1;
		while (true) {
			log_info("[Sync] Start to sync[%d] ...", sync_count);
			// FIXME:  RC_Step1. 订阅RC服务 && 同步
			auto other_gateways = ctx->GatewayCopyer();
			if (other_gateways.empty()) {
				log_warn("[Sync] No other gateways found.");
				goto GATEWAYS_LOOP;
			}

			// 遍历其他网关
			for (const auto &gateway: other_gateways) {
				pair<string, int> ip_port        = parseIpPort(gateway.first);
				auto              target_gateway = create_addr(ip_port.second, ip_port.first.c_str());

				// 遍历本地服务
				for (const auto &service: ctx->LocalServiceCopyer()) {
					// 需要同步的报文
					biz::Message msg     = {.magic = biz::Msg_Type::SGW_SYNCHRO, .flag = 0};
					string       encoded = service.second.encode();
					const char  *data    = encoded.c_str();
					msg.payload_len      = strlen(data) + 1;
					msg.payload_data     = (char *) malloc(msg.payload_len);
					if (msg.payload_data == NULL) {
						log_fatal("[Sync] malloc error: %d", errno);
						free_message(&msg);
						continue;
					}
					memcpy(msg.payload_data, data, msg.payload_len);
					msg.payload_data[msg.payload_len] = '\n';

					// 内存对象
					size_t buffer_size = message_size(msg);
					char  *buffer      = (char *) malloc(buffer_size);
					serialize(&msg, buffer, buffer_size);

					// 创建目标/客户端Socket
					ssize_t sentBytes = sendto(sockfd, buffer, buffer_size, 0, (SOCKADDR *) &target_gateway, sizeof(target_gateway));
					if (sentBytes < 0) {
						log_error("[Sync] sent[FD:%d] error: %d", sockfd, errno);
						free(buffer);
						free_message(&msg);
						continue;
					}
					log_info("[Sync] [PSH] to %s payload { %s }", gateway.first.c_str(), msg.payload_data);
					free(buffer);
					free_message(&msg);
				}
			}

		GATEWAYS_LOOP:
			this_thread::sleep_for(chrono::milliseconds(ctx->config.Discovery.Interval));
			sync_count++;
			printf("\n");
		}
	}

	void SyncModule::Stop() {
		if (!running) return;
		running = false;
		if (thread_sync.joinable()) {
			thread_sync.join();
		}
	}

	SyncModule::~SyncModule() = default;
} // namespace mod
